/**
 * ClassName: HdfsClient <br/>
 * Description: <br/>
 * date: 2021/1/27 9:33<br/>
 *
 * @author Hesion<br />
 * @version
 * @since JDK 1.8
 */

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.Test;
import sun.nio.ch.IOUtil;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

/**
 * @program: ClientDemo
 * @description: 测试HDFSAPI
 * @author: hesion
 * @create: 2021-01-27 09:33
 **/
public class HdfsClient {

    @Test
    public void teskMkdir() throws Exception {
        //1获取文件系统
        Configuration configuration = new Configuration();

        //配置在集群上运行
        FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"),
                configuration, "root");
        // 2 创建目录
        fs.mkdirs(new Path("/test"));
        fs.close();
    }

    @Test
    public void testCopyFromLocal() throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root");
        fs.copyFromLocalFile(new Path("d://lagou.txt"), new Path("/lagou"));
    }

    @Test
    public void testListFiles() throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root");
        RemoteIterator<LocatedFileStatus> listFiles = fs.listFiles(new Path("/"), true);
        while (listFiles.hasNext()) {
            LocatedFileStatus file = listFiles.next();
            System.out.println(file.getPath().getName());
            BlockLocation[] blockLocations = file.getBlockLocations();
            for (BlockLocation blockLocation : blockLocations) {
                String[] hosts = blockLocation.getHosts();
                for (String host : hosts) {
                    System.out.println(host);
                }
            }
        }
        fs.close();
    }

    @Test
    public void testListStatus() throws Exception {
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root");
        FileStatus[] fileStatuses = fs.listStatus(new Path("/"));
        for (FileStatus fileStatus : fileStatuses) {
            if (fileStatus.isDirectory()) {
                System.out.println("d:" + fileStatus.getPath().getName());
            } else {
                System.out.println("f:" + fileStatus.getPath().getName());
            }
        }
        fs.close();
    }

    /**
     * @return
     * @Author hst
     * @Description //TODO IO操作文件上传
     * @Date 13:44 2021/1/27
     * @Param
     **/
    @Test
    public void putFileToHDFS() throws Exception {
        // 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root");

        // 创建输入流 从本地取文件
        FileInputStream in = new FileInputStream(new File("d:/lagou_put_test.txt"));

        // 创建输出流，输出到文件HDFS
        FSDataOutputStream out = fs.create(new Path("/lagou_put_test.txt"));
        IOUtils.copyBytes(in, out, configuration);
        IOUtils.closeStream(in);
        IOUtils.closeStream(out);
        fs.close();
    }

    @Test
    public void getFileFromDFS() throws Exception {
        // 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root");

        // 获取文件输入流
        FSDataInputStream fis = fs.open(new Path("/lagou_put_test.txt"));

        // 获取输出流
        FileOutputStream fos = new FileOutputStream(new File("d:/get_from_hdfs.txt"));

        //流的对拷
        IOUtils.copyBytes(fis, fos, configuration);
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);

        fs.close();
    }

    @Test
    public void readSeekFile2() throws Exception {
        // 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root");

        FSDataInputStream in = null;
        try {
            in = fs.open(new Path("/lagou_put_test.txt"));
            IOUtils.copyBytes(in, System.out, 4096);
            in.seek(0);
            IOUtils.copyBytes(in, System.out, 4096);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Test
    public void testUploadPacket() throws Exception {

        // 获取文件系统
        Configuration configuration = new Configuration();
        FileSystem fs = FileSystem.get(new URI("hdfs://linux121:9000"), configuration, "root");

        // 准备本地
        FileInputStream in = new FileInputStream(new File("d:/test_emp_pk.txt"));

        //2 准备好写出数据到hdfs的输出流
        FSDataOutputStream out = fs.create(new Path("/testPacket.txt"), new Progressable() {
            @Override
            public void progress() {
                System.out.println("&");
            }
        });

        //流对拷
        IOUtils.copyBytes(in,out,configuration);
        fs.close();
    }
}

